package com.yanggu.flink.tableapi_sql.define_table

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala._

//使用SQL的DDL语句来定义表, 最常用的方式
//可以定义watermark字段
object SQLForTable {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment

    val settings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()

    val tableEnvironment = StreamTableEnvironment.create(environment, settings)

    //在TableEnv中注册表
    tableEnvironment.executeSql(
      s"""
         |CREATE TABLE sensor(
         | id STRING,
         | `timestamp` BIGINT,
         | temperature DOUBLE
         |) WITH (
         |  'connector'='filesystem',
         |  'path'='file://${getClass.getResource("/sensor.txt").getPath}',
         |  'format'='csv'
         |)
         |
         |""".stripMargin)

    tableEnvironment.sqlQuery(
      """
        |SELECT
        |   id,
        |-- 这里需要使用``符号进行转义, timestamp是SQL中的关键字
        |   MAX(`timestamp`),
        |   MAX(temperature)
        |FROM
        |   sensor
        |GROUP BY
        |   id
        |""".stripMargin)
      .execute()
      .print()

  }

}
